package com.company.cloud.common.mq.redis.client.factory;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StreamOperations;

import com.company.cloud.common.mq.redis.client.consumer.ISaasMqConsumer;
import com.company.cloud.common.mq.redis.client.consumer.annotation.MQRedisConsumer;
import com.company.cloud.common.mq.redis.client.consumer.thread.ConsumerRedisThread;

import cn.hutool.core.util.IdUtil;

public class RedisMQClientFactory {

	private final static Logger logger = LoggerFactory.getLogger(RedisMQClientFactory.class);

	private List<ISaasMqConsumer> consumerList;

	private StreamOperations<String, String, String> streamOperations;

	public void setConsumerList(List<ISaasMqConsumer> consumerList) {
		this.consumerList = consumerList;
	}

	public void setStreamOperations(StreamOperations<String, String, String> streamOperations) {
		this.streamOperations = streamOperations;
	}

	public void init() {
		this.validConsumer();
		this.startConsumer();
	}

	public void destroy() {
		this.destoryClientFactoryThreadPool();
	}

	// ---------------------- thread pool ----------------------

	private ExecutorService clientFactoryThreadPool = Executors.newCachedThreadPool();

	public static volatile boolean clientFactoryPoolStoped = false;

	/**
	 * destory consumer thread
	 */
	private void destoryClientFactoryThreadPool() {
		clientFactoryPoolStoped = true;
		clientFactoryThreadPool.shutdownNow();
	}

	// ---------------------- queue consumer ----------------------
	private List<ConsumerRedisThread> consumerRespository = new ArrayList<ConsumerRedisThread>();

	/***/
	@SuppressWarnings({ "unchecked", "rawtypes" })
	private void validConsumer() {
		if (this.streamOperations == null) {
			logger.warn("+++++ MQ-Redis, Redis  not config.");
			return;
		}

		if (this.consumerList == null || this.consumerList.size() == 0) {
			logger.warn("+++++ MQ-Redis, MQConsumer not found.");
			return;
		}

		for (ISaasMqConsumer consumer : this.consumerList) {
			MQRedisConsumer annotation = consumer.getClass().getAnnotation(MQRedisConsumer.class);
			if (annotation == null) {
				throw new RuntimeException("+++++ MQ-Redis, MQConsumer(" + consumer.getClass() + "), annotation is not exists.");
			}
			if (annotation.group() == null || annotation.group().trim().length() == 0) {
				try {
					InvocationHandler invocationHandler = Proxy.getInvocationHandler(annotation);
					Field mValField = invocationHandler.getClass().getDeclaredField("memberValues");
					mValField.setAccessible(true);
					Map memberValues = (Map) mValField.get(invocationHandler);
					String randomGroup = IdUtil.fastSimpleUUID();
					memberValues.put("group", randomGroup);
				} catch (Exception e) {
					throw new RuntimeException("+++++ MQ-Redis, MQConsumer(" + consumer.getClass() + "), group empty and genereta error.");
				}
			}

			if (annotation.group() == null || annotation.group().trim().length() == 0) {
				throw new RuntimeException("+++++ MQ-Redis, MQConsumer(" + consumer.getClass() + "),group is empty.");
			}

			// valid topic
			if (annotation.topic() == null || annotation.topic().trim().length() == 0) {
				throw new RuntimeException("+++++ MQ-Redis, MqConsumer(" + consumer.getClass() + "), topic is empty.");
			}

			// consumer map
			consumerRespository.add(new ConsumerRedisThread(consumer, this.streamOperations));
		}
	}

	/***/
	private void startConsumer() {
		if (this.consumerRespository == null || this.consumerRespository.size() == 0) {
			return;
		}
		// execute thread
		for (ConsumerRedisThread item : consumerRespository) {
			clientFactoryThreadPool.execute(item);
			logger.info("+++++ MQ-Redis, consumer init success, , topic:{}, group:{}", item.getMqConsumer().topic(),item.getMqConsumer().group());
		}
	}

}
